// Copyright 2014 mqant Author. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//	http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package rpc

import (
	"context"
	"fmt"

	"gitee.com/lakertt/newtbig/common"
	log "gitee.com/lakertt/newtbig/logging"
	"gitee.com/lakertt/newtbig/module"
	"gitee.com/lakertt/newtbig/msg/framepb"
	mq "gitee.com/lakertt/newtbig/rpc/mq"
	"gitee.com/lakertt/newtbig/utils"
)

type RPCClient struct {
	app         module.App
	nats_client *mq.NatsClient
	opts        module.Options
}

func NewRPCClient(app module.App) (module.RPC, error) {
	rpc_client := new(RPCClient)
	rpc_client.app = app
	rpc_client.opts = app.Options()
	nats_client, err := mq.NewNatsClient(app)
	if err != nil {
		log.Error("NewRPCClient err: %s", err.Error())
		return nil, err
	}
	rpc_client.nats_client = nats_client
	return rpc_client, nil
}

func (c *RPCClient) CallBack() chan *framepb.Msg {
	return c.nats_client.CallBack()
}

func (c *RPCClient) Start() (err error) {
	utils.SafeGO(func() {
		c.nats_client.StartPublishHandle()
	})
	utils.SafeGO(func() {
		c.nats_client.StartRequestHandle()
	})

	return nil
}

func (c *RPCClient) Stop() (err error) {
	if c.nats_client != nil {
		c.nats_client.Done()
	}
	return nil
}
func (c *RPCClient) route(msg *framepb.Msg) (string, error) {
	dis := c.app.GetDispatcher()
	if dis == nil {
		return "", fmt.Errorf("rpc getdispatcher is nil err")
	}
	route, err := dis.Route(msg)
	if err != nil {
		return "", err
	}
	return route, nil
}

func (c *RPCClient) rRoute(msg *framepb.Msg) (string, error) {
	route, err := c.route(msg)
	if err != nil {
		return "", err
	}
	return fmt.Sprintf("%s_req", route), nil
}

func (c *RPCClient) bRoute(msg *framepb.Msg) (string, error) {
	route, err := c.route(msg)
	if err != nil {
		return "", err
	}
	return fmt.Sprintf("%s.%s", route, common.BROATCAST), nil
}

func (c *RPCClient) pRoute(msg *framepb.Msg) (string, error) {
	route, err := c.route(msg)
	if err != nil {
		return "", err
	}
	return fmt.Sprintf("%s.%s", route, "pub"), nil
}

func (c *RPCClient) sRoute(msg *framepb.Msg) (string, error) {
	return fmt.Sprintf("%s.%s.%s", c.opts.AppName, c.opts.ServerType, c.opts.ServerName), nil
}

func (c *RPCClient) Call(ctx context.Context, msg *framepb.Msg) (*framepb.Msg, error) {
	log.Logger.Debugf("rpc call seq:%d cmd:%d uid:%d  key:%d  data:%s", msg.Seq, msg.ID, msg.UID, msg.Key, msg.Body)
	rout, err := c.rRoute(msg)
	if err != nil {
		utils.Put(msg)
		return nil, err
	}
	log.Logger.Debugf("rpc call rout:%s ", rout)
	r, err := c.nats_client.RequestMsg(rout, msg)
	return r, err
}

func (c *RPCClient) CallNR(msg *framepb.Msg) (err error) {
	log.Logger.Debugf("rpc cn seq:%d cmd:%d uid:%d  key:%d  data:%s", msg.Seq, msg.ID, msg.UID, msg.Key, msg.Body)
	rout, err := c.pRoute(msg)
	if err != nil {
		utils.Put(msg)
		return err
	}
	log.Logger.Debugf("rpc callnr rout:%s ", rout)
	return c.nats_client.PublishMsg(rout, msg)
}

func (c *RPCClient) CallB(msg *framepb.Msg) (err error) {
	log.Logger.Debugf("rpc cb seq:%d cmd:%d uid:%d  key:%d  data:%s", msg.Seq, msg.ID, msg.UID, msg.Key, msg.Body)
	rout, err := c.bRoute(msg)
	if err != nil {
		utils.Put(msg)
		return err
	}
	log.Logger.Debugf("rpc callb rout:%s ", rout)
	return c.nats_client.PublishMsg(rout, msg)
}

func (c *RPCClient) CallS(msg *framepb.Msg) (err error) {
	log.Logger.Debugf("rpc s seq:%d cmd:%d uid:%d  key:%d  data:%s", msg.Seq, msg.ID, msg.UID, msg.Key, msg.Body)
	rout, err := c.sRoute(msg)
	if err != nil {
		utils.Put(msg)
		return err
	}
	log.Logger.Debugf("rpc calls rout:%s ", rout)
	err = c.nats_client.SendMsg(rout, msg)
	return err
}
